package com.example.kafka.consumer;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;


@Slf4j
@Component
public class Consumer {
	
	@KafkaListener(topics = "helloworld.t")
    public void receiveMessage(ConsumerRecord<Integer,String>  record) {
		
		log.info("ConsumerRecord:key:{},value:{},partition:{},offset:{}",
				record.key(),record.value(),record.partition(),record.offset());
	}
	
	@KafkaListener(topics = "helloworld.t2",containerFactory = "kafkaBatchListenerContainerFactory")
    public void receiveMessage(List<ConsumerRecord<Integer,String>>  records, Acknowledgment ack) {
		log.info("Size:{}",records.size());
		for(ConsumerRecord<Integer,String> record : records) {
			log.info("ConsumerBatchRecord:key:{},value:{},partition:{},offset:{}",
					record.key(),record.value(),record.partition(),record.offset());
		}
		ack.acknowledge();//手动提交偏移量
	}
}
